package com.lagouedu.order.consumer;

import com.alibaba.fastjson.JSONObject;
import com.lagouedu.order.entity.Inventory;
import com.lagouedu.order.repository.InventoryRepository;
import com.lagouedu.order.util.Constants;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Component
@Slf4j
public class MessageListener implements MessageListenerConcurrently {

    @Autowired
    private InventoryRepository inventoryRepository;
    @Autowired
    private TransactionMQProducer transactionMQProducer;
    @Autowired
    private RedissonClient redissonClient;

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        MessageExt ext = list.get(0);
        // 处理消息
        String msg = new String(ext.getBody());
        JSONObject jsonObject = JSONObject.parseObject(msg);
        // 减库存 并且发送消息给订单服务
        if(doReduceProductAndSendMsg(jsonObject)){
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
    }


    public boolean doReduceProductAndSendMsg(JSONObject msg){
        TransactionListener transactionListener = new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                // 加锁
                RLock lock = redissonClient.getLock("lock:dbinventory");
                lock.lock();
                // 数据库减库存
                reduceDbInventory();
                lock.unlock();

                return LocalTransactionState.COMMIT_MESSAGE;
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                // 回调检查是否扣减库存

                return LocalTransactionState.COMMIT_MESSAGE;
            }
        };
        ExecutorService executorService= Executors.newFixedThreadPool(10);
        transactionMQProducer.setExecutorService(executorService);
        transactionMQProducer.setTransactionListener(transactionListener);
        Message message = new Message(Constants.INVENTORY_REDUCE_TOPIC, msg.toJSONString().getBytes());
        try {
            transactionMQProducer.sendMessageInTransaction(message, msg.get("userId"));
            log.info("发送消息成功--topic ->{}--msg -> {}", Constants.INVENTORY_REDUCE_TOPIC, msg);
        } catch (MQClientException e) {
            e.printStackTrace();
            return false;
        }
        return true;
    }

    @Transactional
    public void reduceDbInventory(){
        // 数据库减库存
        Inventory inventory = inventoryRepository.getOne(1);
        inventory.setRemainNum(inventory.getRemainNum() - 1);
        inventoryRepository.save(inventory);
        log.info("数据库减少库存---{}", inventory.getRemainNum());
    }
}
